package com.intsig.airflow.web.rest;

import com.intsig.airflow.web.ajax.RestResponseResult;
import com.intsig.airflow.web.dto.DagEntityDto;
import com.intsig.airflow.web.model.DagEdge;
import com.intsig.airflow.web.model.DagEntity;
import com.intsig.airflow.web.model.DagNode;
import com.intsig.airflow.web.test.MockDagValue;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.*;

import java.util.*;
import java.util.stream.Collectors;

/**
 * @author leojie 2021/3/13 1:39 下午
 */
@RequestMapping("/airflow/dag")
@RestController
public class DagController {

    @GetMapping(value = "/test")
    public RestResponseResult<String> testGetRequest(@RequestParam String name){
        String response = "hello "+name;
        return RestResponseResult.ok(response).message("后端处理数据成功");
    }

    @GetMapping(value = "/list")
    public RestResponseResult<List<DagEntityDto>> findAllDags() {
        DagEntity dagEntity = MockDagValue.mockDag();
        return RestResponseResult.ok(Collections.singletonList(new DagEntityDto().convertFor(dagEntity)));
    }

    @PostMapping("/save")
    public RestResponseResult saveDag(@RequestBody @Validated DagEntityDto dagEntityDto) {
        DagEntity dagEntity = dagEntityDto.convertTo();
        return RestResponseResult.ok().message("DAG 保存成功");
    }

    @PostMapping("/generateTaskDependency")
    public RestResponseResult<Map<String, String>> generateTaskDependency(@RequestBody @Validated DagEntityDto dagEntityDto) {
        String defaultTaskDependency = "%s >> %s";

        DagEntity dagEntity = dagEntityDto.convertTo();
        final List<DagEdge> edges = dagEntity.getDagEdges();

        final List<DagNode> nodes = dagEntity.getDagNodes();
        Map<Long, String> nodeMap = nodes.stream().collect(Collectors.toMap(DagNode::getId, DagNode::getName));
        StringBuilder dependencyBuilder = new StringBuilder();
        edges.forEach(edge -> {
            String taskDependency = String.format(defaultTaskDependency, nodeMap.get(edge.getSrcNodeId()), nodeMap.get(edge.getDstNodeId()));
            dependencyBuilder.append(taskDependency).append("\n");
        });
        Map<String,String> dependencyResMap = new HashMap<>(1);
        dependencyResMap.put("taskDependencies", dependencyBuilder.toString());

        return RestResponseResult.ok(dependencyResMap).message("airflow task 依赖配置生成成功");
    }

    @PostMapping("/add")
    public RestResponseResult<DagEntity> addDag(DagEntityDto dagEntityDto) {
        DagEntity defaultDag = createDefaultDag();
        return RestResponseResult.ok(defaultDag).message("DAG 新增成功");
    }

    private DagEntity createDefaultDag() {
        DagEntity dagEntity = new DagEntity();

        Map<String, String> nodeStyle = new HashMap<>(1);
        nodeStyle.put("border", "1px solid greenYellow");
        dagEntity.setLabel("test_dag");

        DagEdge edge = new DagEdge();
        edge.setId(1L);
        edge.setSrcNodeId(1L);
        edge.setSrcOutputIdx(0);
        edge.setDstNodeId(2L);
        edge.setDstInputIdx(0);
        dagEntity.addEdges(Collections.singletonList(edge));


        DagNode node1 = new DagNode();
        node1.setId(1L);
        node1.setName("start");
        node1.setInPorts(new int[]{0});
        node1.setOutPorts(new int[]{0});
        node1.setPosX(157f);
        node1.setPosY(120f);
        node1.setType("Variable");
        node1.setNodeStyle(nodeStyle);
        node1.addNodeRightClickEvent("do some thing1", "eventOfDoSomeThingFirst");
        node1.addNodeRightClickEvent("do some thing2", "eventOfDoSomeThingSecond");

        DagNode node2 = new DagNode();
        node2.setId(2L);
        node2.setName("end");
        node2.setInPorts(new int[]{0});
        node2.setOutPorts(new int[]{0});
        node2.setPosX(459f);
        node2.setPosY(257f);
        node2.setType("Variable");
        node2.setNodeStyle(nodeStyle);
        node2.addNodeRightClickEvent("do some thing1", "eventOfDoSomeThingFirst");
        node2.addNodeRightClickEvent("do some thing2", "eventOfDoSomeThingSecond");

        dagEntity.addNodes(Arrays.asList(node1, node2));

        return dagEntity;
    }
}
